package com.yzh.dxpro.p11.taskEleven;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.yzh.dxpro.p4.utils.TProperties;

public class ElevenReduce extends Reducer<Text,Text,NullWritable,Text>{
	private int cyc;
	@Override
	protected void setup(Reducer<Text, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		cyc=3;
	}
	
	Text result=new Text();
	@Override
	protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		int liveDays=0;
		int joinDays=0;
		int not=0;
		int flag=2;
		int doflag=0;
		Map<String,String> map=new HashMap();
		String temp="";
		String liveStr="";
		for (Text val:values) {
			String[] str=val.toString().split(TProperties.getValue("fileoutsplit"));
			if("0".equals(str[0])) {
				flag=Integer.parseInt(str[4]);
				doflag=doflag+1;
				liveStr=val.toString();
				map.put("1",str[5]);
			}else if("1".equals(str[0])) {
				doflag=doflag+2;
				map.put("1", str[1]);
			}
		}
		String[] res=liveStr.split(TProperties.getValue("fileoutsplit"));
		switch(doflag) {
		case 1:
			temp=map.get("0");
			if(flag==1) {
				liveDays=Integer.parseInt(res[1]);
				joinDays=Integer.parseInt(res[2]);
				not=Integer.parseInt(res[3]);
			}else {
				joinDays=Integer.parseInt(res[2])+1;
				not=Integer.parseInt(res[3])+1;
				if(joinDays<cyc) {
					liveDays=Integer.parseInt(res[1])+1;
					flag=2;
				}else {
					if(not>=cyc) {
						liveDays=Integer.parseInt(res[1]);
						flag=1;
					}else {
						liveDays=Integer.parseInt(res[1])+1;
						flag=0;
					}
				}
			}
			break;
		case 2:
			temp=map.get("1");
			liveDays=1;
			joinDays=1;
			not=0;
			flag=2;
			break;
		case 3:
			temp=map.get("1");
			if(flag==1) {
				liveDays=Integer.parseInt(res[1]);
				joinDays=Integer.parseInt(res[2]);
				not=Integer.parseInt(res[3]);
				flag=1;
			}else {
				joinDays=Integer.parseInt(res[2])+1;
				liveDays=Integer.parseInt(res[1])+1;
				not=0;
				if(joinDays<cyc) 
				{
					flag=2;
				}else {
					flag=0;
				}
			}
			break;
		}
		StringBuffer sb=new StringBuffer();
		sb.append(key.toString()).append(TProperties.getValue("outfilesplit")).append(liveDays)
		.append(TProperties.getValue("outfilesplit")).append(joinDays)
		.append(TProperties.getValue("outfilesplit")).append(not)
		.append(TProperties.getValue("outfilesplit")).append(flag)
		.append(TProperties.getValue("outfilesplit")).append(temp);
		result=new Text(sb.toString());
		context.write(NullWritable.get(), result);
	}

	
}


